package com.guli.order.config;

import com.guli.common.constant.MessageQueueConstant;
import com.guli.common.constant.OrderConstant;
import com.guli.common.constant.WareConstant;
import com.guli.common.to.StockWareTo;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.impl.AMQImpl;
import lombok.extern.slf4j.Slf4j;
import lombok.extern.slf4j.XSlf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

/*
 * @Author 罗俊
 * @date 2020/11/23 - 10:52 下午
 *
 */
@Component
@Slf4j
public class RabbitMqConfig implements InitializingBean {

    @Resource
    AmqpAdmin amqpAdmin;


    /**
     * 延迟队列
     *
     * @return
     */
//    @Bean
    public Queue userOrderDelayQueue() {

        // 设置延迟队列的参数
        Map<String, Object> arguments = new HashMap<>();

        arguments.put(MessageQueueConstant.X_DEAD_LETTER_EXCHANGE, OrderConstant.ORDER_EVENT_EXCHANGE);

        arguments.put(MessageQueueConstant.X_DEAD_LETTER_ROUTING_KEY, OrderConstant.ORDER_RELEASE_ORDER_ROUTING_KEY);

        // 消息过期时间（单位：毫秒）
        arguments.put(MessageQueueConstant.X_MESSAGE_TTL,/*60*1000*/OrderConstant.ORDER_PAYMENT_TIME); // 30分钟后仍然未支付，就取消订单

        return new Queue(OrderConstant.ORDER_DELAY_QUEUE, true, false, false, arguments);
    }

    /**
     * 其他服务订阅的队列
     *
     * @return
     */
//    @Bean
    public Queue userOrderQueue() {

        Map<String, Object> arguments = new HashMap<>();

        return new Queue(OrderConstant.ORDER_RELEASE_QUEUE, true, false, false, arguments);
    }

    /**
     * Dead Letter Exchange （死信交换机） + 普通交换机
     *
     * @return
     */
//    @Bean
    public Exchange userOrderDelayExchange() {

        log.info("创建{}，交换机", OrderConstant.ORDER_EVENT_EXCHANGE);

        return new TopicExchange(OrderConstant.ORDER_EVENT_EXCHANGE, true, false);
    }

    /**
     * 绑定 消息从交换机发送到延时队列的路由
     *
     * @return
     */
//    @Bean
    public Binding orderCreateOrder() {
        //Binding(String destination, DestinationType destinationType, String exchange, String routingKey,
        //			@Nullable Map<String, Object> arguments)
        Map<String, Object> arguments = new HashMap<>();

        return new Binding(OrderConstant.ORDER_DELAY_QUEUE, Binding.DestinationType.QUEUE, OrderConstant.ORDER_EVENT_EXCHANGE, OrderConstant.ORDER_CREATE_ORDER_ROUTING_KEY, arguments);
    }

    /**
     * 无需绑定 消息从延时队列到交换机的路由，因为延迟队列中的参数已经指定了
     *
     * @return
     */
/*    @Bean
    public Binding orderReleaseOrderToExchange() {
        //Binding(String destination, DestinationType destinationType, String exchange, String routingKey,
        //			@Nullable Map<String, Object> arguments)
        Map<String, Object> arguments = new HashMap<>();

        return new Binding(ORDER_EVENT_EXCHANGE, Binding.DestinationType.EXCHANGE, ORDER_DELAY_QUEUE, ORDER_RELEASE_ORDER_ROUTING_KEY, arguments);
    }*/

    /**
     * 绑定 消息从交换机到最终队列的路由
     *
     * @return
     */
//    @Bean
    public Binding orderReleaseOrderToQueue() {
        //Binding(String destination, DestinationType destinationType, String exchange, String routingKey,
        //			@Nullable Map<String, Object> arguments)
        Map<String, Object> arguments = new HashMap<>();

        return new Binding(OrderConstant.ORDER_RELEASE_QUEUE, Binding.DestinationType.QUEUE, OrderConstant.ORDER_EVENT_EXCHANGE, OrderConstant.ORDER_RELEASE_ORDER_ROUTING_KEY, arguments);
    }


    /**
     * 使用 AmqpAdmin 对交换机和队列等进行操作会覆盖，
     * 但是使用 @Bean 直接创建的不能覆盖原有的交换机、队列等（只有在第一次监听队列的时候才会创建队列，交换机，绑定）
     *
     * @throws Exception
     */
//    @PostConstruct // 构造器执行之后就运行这个方法
    @Override
    public void afterPropertiesSet() throws Exception {

        amqpAdmin.declareExchange(userOrderDelayExchange());

        amqpAdmin.declareQueue(userOrderDelayQueue());

        amqpAdmin.declareQueue(userOrderQueue());

        /**
         * 交换机和队列都创建完成之后才能 进行绑定
         * */
        amqpAdmin.declareBinding(orderReleaseOrderToQueue());

        amqpAdmin.declareBinding(orderCreateOrder());

        /**
         * order.event.exchange 订单交换机到 stock.release.queue 库存队列的路由键
         * 用于在订单更新之后，立即通知库存服务进行修改（库存解锁等）
         * */
        Binding updateOrderAndUpdateWareBinding = new Binding(WareConstant.STOCK_EVENT_EXCHANGE, Binding.DestinationType.EXCHANGE, OrderConstant.ORDER_EVENT_EXCHANGE, WareConstant.STOCK_RELEASE_CANCEL_ORDER_ROUTING_KEY, null);

        amqpAdmin.declareBinding(updateOrderAndUpdateWareBinding);

        /**
         * 接收秒杀服务发送过来的订单信息，并执行创建订单扣库存的操作
         * */
        Queue SecKillCreateOrderQueue = new Queue(OrderConstant.SEC_KILL_CREATE_ORDER_QUEUE, true, false, false, null);

        amqpAdmin.declareQueue(SecKillCreateOrderQueue);

        Binding secKillCreateOrderBinding = new Binding(OrderConstant.SEC_KILL_CREATE_ORDER_QUEUE, Binding.DestinationType.QUEUE, OrderConstant.ORDER_EVENT_EXCHANGE, OrderConstant.SEC_KILL_CREATE_ORDER_ROUTING_KEY, null);

        amqpAdmin.declareBinding(secKillCreateOrderBinding);

        this.rabbitTemplateConfig();
    }


    /**
     * 测试消息队列的接受者
     *
     * @param stockWareTo
     */
//    @RabbitListener(queues = {ORDER_RELEASE_ORDER_QUEUE})
    public void receiveOrder(StockWareTo stockWareTo, Channel channel,Message message) {

        System.out.println("测试消息接受者，接收到订单消息：" + stockWareTo + "\t" + new Date());
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        try {
            channel.basicAck(deliveryTag,false);
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    @Resource
    RabbitTemplate template;

    public void rabbitTemplateConfig() {
//        RabbitTemplate template = new RabbitTemplate();
//        configurer.configure(template, connectionFactory); 因为 spring boot 2.3 之后才有这个类，所以暂时用不了

        log.info("正在设置 RabbitMQ 发送消息回报");

        // 1、服务器（RabbitMQ）收到消息就回调
        RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {

            /**
             *
             * @param correlationData 当前消息的唯一关联数据（这个是消息的唯一id）
             * @param ack   消息是否成功
             * @param cause 失败的原因
             */
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {

                System.out.println("MQ已经收到消息，消息的唯一id（需要在发送消息时指定）correlationData = " + correlationData + "\tack = " + ack + "\tcause = " + cause);

                //TODO 做好消息确认机制（publisher,consumer 【手动ack】）
                //TODO 每一个发送的消息都在数据库做好记录，定期将失败的消息再发送一遍
            }
        };

        template.setConfirmCallback(confirmCallback);

        // 2、消息抵达队列的确认回调 （spring boot 2.3 之后）
        /*RabbitTemplate.ReturnsCallback returnsCallback = new RabbitTemplate.ReturnsCallback(){

         *//**
         *  只要消息队列没有投递给指定的队列，就触发这个失败回调
         * @param returned
         *//*
            @Override
            public void returnedMessage(ReturnedMessage returned) {

                Message message = returned.getMessage(); // 投递失败的消息的相信信息

                int replyCode = returned.getReplyCode(); // 回复的状态码

                String replyText = returned.getReplyText(); // 回复的文本内容

                String exchange = returned.getExchange(); // 当时这个消息发送给哪一个交换机

                String routingKey = returned.getRoutingKey(); // 当时这个消息发送给哪一个交换机


                log.error("发送到交换机【{}】失败，此次发送的消息是【{}】，返回码是【{}】，返回的文本内容是【{}】",exchange,message,replyCode,replyText);
            }
        };*/

//        template.setReturnsCallback(returnsCallback); spring boot 2.3 以后的命名方式


        RabbitTemplate.ReturnCallback returnCallback = new RabbitTemplate.ReturnCallback() {

            /**
             *
             * @param message 投递失败的消息的相信信息
             * @param replyCode 回复的状态码
             * @param replyText 回复的文本内容
             * @param exchange 当时这个消息发送给哪一个交换机
             * @param routingKey 当时这个消息发送给哪一个交换机
             */
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {


                log.error("发送到交换机【{}】失败，此次发送的消息是【{}】，返回码是【{}】，返回的文本内容是【{}】", exchange, message, replyCode, replyText);

            }
        };

        log.info("正在配置 消息抵达队列的确认回调");

        template.setReturnCallback(returnCallback);

    }
}
